package com.rxjava.rxjava;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

import org.junit.Test;

import rx.Observable;
import rx.Scheduler;
import rx.Subscriber;
import rx.exceptions.OnErrorThrowable.OnNextValue;
import rx.observables.StringObservable;
import rx.schedulers.Schedulers;
import rx.subjects.BehaviorSubject;

public class RxJavaDemo {
	
	private String str;
	BehaviorSubject<Object> subject = BehaviorSubject.create("default");
	


	Observable<List<String>> query(){
		String[] strs = new String[]{"url2","url3","url4"};
		return Observable.create(
				(f) -> {
					f.onNext(Arrays.asList(strs));
					f.onCompleted();
				}
//				f -> f.onNext(Arrays.asList(strs))
		);
	}

	@Test
	public void test() {
		Observable.just("Hello, world!")
	    .map(s -> s.hashCode())
	    .map(i -> Integer.toString(i))
	    .subscribe(s -> System.out.println(s));
	}
	
	@Test
	public void test2(){
		Observable.just("Hello, world!")
	    .map(s -> s + " -Dan")
	    .map(s -> s.hashCode())
	    .map(i -> Integer.toString(i))
	    .subscribe(s -> System.out.println(s),error-> System.out.println(error),()-> System.out.print("finish"));
	}
	
	@Test
	public void test3(){
		Observable.just(Arrays.asList("url2","url3","url4")).flatMap(urls -> Observable.from(urls)).subscribe(url -> System.out.println(url));
//		Observable.create(f -> Arrays.asList(strs)).subscribe(url -> System.out.println(url));
//		this.query().flatMap(urls -> Observable.from(urls))  
//		//.filter(s -> s.equals("url2"))
//	    .subscribe(url -> System.out.println(url));  
	}
	
	public Observable<String> getTitle(String url){
		return Observable.just(url);
	}
	
	@Test
	public void test5(){
		this.query().flatMap(urls -> Observable.from(urls))
		.flatMap(s-> getTitle(s))
		.filter(s -> s.equals("url2"))
	    .subscribe(url -> {System.out.println(url);},error->System.out.println(error),()->System.out.println("complete"));  
	}
	
	@Test
	public void test4(){
		Observable<String> obs = Observable.create(
			new Observable.OnSubscribe<String>() {  
		        @Override  
		        public void call(Subscriber<? super String> sub) {  
		            sub.onNext("Hello, world!");  
		            sub.onCompleted();  
		        }  
	    }); 
		
		obs.subscribe(s-> System.out.println(s));
		
	}
	
	@Test
	public void test6(){
		Observable.from(Arrays.asList("s2","s3","s5")).filter(s -> !s.equals("s3"))
	    .subscribe(url -> {System.out.println(url);},error->System.out.println(error),()->System.out.println("complete"));  
	}
	
	@Test
	public void test7(){
		Observable.just("Hello", "Hi", "Aloha").subscribe(s->System.out.println(s));
	}
	
	@Test
	public void test8(){
		Observable.just("hello").subscribeOn(Schedulers.io()).subscribe(s->System.out.println(s));
	}
	
	@Test
	public void test9(){
		StringObservable.encode(Observable.just("hello world"), "utf-8").subscribe(s->System.out.println(s));
	}

	@Test
	public void test10(){
		StringObservable.decode(Observable.just("hello world".getBytes()), "utf-8").subscribe(s->System.out.println(s));
	}
	
	@Test
	public void test11(){
		StringObservable.stringConcat(Observable.just("hello","aaa","bbb")).subscribe((s)-> {System.out.println(s);});
	}
	
	@Test
	public void test12(){
		StringObservable.split(Observable.just("hello123"),"(\\d+)").subscribe(s->System.out.println(s));
	}
	
	public Observable<String> get(){
		return StringObservable.stringConcat(Observable.just("hello","aaa","bbb"));
	}
	
	@Test
	public void test13(){
		Observable<String> obs = StringObservable.stringConcat(Observable.just("hello","aaa","bbb"));
		obs.subscribe((s)->{str=s;});
		System.out.println();
	}
	
	@Test
	public void test14() throws InterruptedException{
		Observable.just("helo","fdsf","adfa")
			.subscribeOn(Schedulers.newThread())
			.subscribe((s)->{
				System.out.println(s);
				System.out.println(Thread.currentThread().getId());
				System.out.println(Thread.currentThread().getName());
			});
		
		Thread.sleep(2000);
	}
	
	
}
